package com.pk.flink.functions.sink;

import com.pk.flink.bean.Access;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

public class AccessConsoleSinkFunction extends RichSinkFunction<Access> {
    private int subTaskIndex;
    @Override
    public void invoke(Access access, Context context) throws Exception {

        System.out.println("Custom Sink：" + subTaskIndex + " > " + access);
    }

    @Override
    public void close() throws Exception {
        super.close();
    }

    @Override
    public void open(org.apache.flink.configuration.Configuration parameters) throws Exception {
        super.open(parameters);
        RuntimeContext runtimeContext = getRuntimeContext();
        subTaskIndex = runtimeContext.getIndexOfThisSubtask();
    }
}
